import org.apache.spark.sql.ForeachWriter

import java.sql.{Connection, DriverManager, PreparedStatement}

class MySQLWriter extends ForeachWriter[BusInfo] {

  var conn: Connection = _
  var stat: PreparedStatement = _

  //获取连接
  override def open(partitionId: Long, epochId: Long): Boolean = {
    if (conn == null){
      conn = MySQLWriter.getConnection
    }
    true
  }

  //处理数据
  override def process(value: BusInfo): Unit = {
    //获得经纬度
    val arr: Array[String] = value.lglat.split("_")
    //数据插入car_gps表
    val sql = "insert into car_gps(deployNum,plateNum,timeStr,lng,lat) value(?,?,?,?,?)"
    stat = conn.prepareStatement(sql)
    stat.setString(1, value.deployNum)
    stat.setString(2, value.plateNum)
    stat.setString(3, value.timeStr)
    stat.setString(4, arr(0))
    stat.setString(5, arr(1))
    stat.executeUpdate()
  }

  //释放连接
  override def close(errorOrNull: Throwable): Unit = {
    if (conn != null)
      conn.close()
    if (stat != null)
      stat.close()
  }
}

object MySQLWriter {

  //初始化连接
  var conn: Connection = _
  val url = "jdbc:mysql://linux123:3306/lg_logstic?useUnicode=true&characterEncoding=utf8&useSSL=false"
  val user = "root"
  val pwd = "12345678"
  def getConnection: Connection = {
    if (conn == null || conn.isClosed){
      conn = DriverManager.getConnection(url, user, pwd)
    }
    conn
  }

}